package main

import (
	"DMSKafka/config"
	"context"
	"github.com/Shopify/sarama"
	"log"
)

// 消费者组
func main() {
	//获取kafka参数
	conf, err := config.GetConfig()
	if err != nil {
		log.Printf("fail to start consumer, err:%v\n", err)
		return
	}
	//初始化消费者组
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	//创建消费者组
	group, err := sarama.NewConsumerGroup(conf.Host, "mygroup", config)
	if err != nil {
		log.Printf("%+v\n", err)
		return
	}
	defer group.Close()
	go func() {
		for err := range group.Errors() {
			log.Printf("%+v\n", err)
		}
	}()
	ctx := context.Background()
	for {
		topic := []string{conf.Topic}
		handler := Consumer{}
		err := group.Consume(ctx, topic, handler)
		if err != nil {
			log.Printf("%+v\n", err)
			return
		}
	}

}

type Consumer struct {
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (Consumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
		session.MarkMessage(message, "")
	}

	return nil
}
